package com.raven.play.stream.test.javadsl.tutorial;

import akka.Done;
import akka.NotUsed;
import akka.japi.Pair;
import akka.stream.*;
import akka.stream.javadsl.*;
import com.raven.play.stream.test.BaseTest;
import org.junit.jupiter.api.Test;

import java.util.concurrent.CompletionStage;

public class $09GraphBasics extends BaseTest {

    @SuppressWarnings("unchecked")
    @Test
    public void test(){

        RunnableGraph<NotUsed> graph = RunnableGraph.fromGraph(GraphDSL.create(builder -> {

            Source<Integer, NotUsed> input = Source.range(1, 10);
            Flow<Integer, Integer, NotUsed> incrementer = Flow.of(Integer.class).map(it -> it + 1);
            Flow<Integer, Integer, NotUsed> multiplier = Flow.of(Integer.class).map(it -> it * 2);
            Sink<Pair<Integer, Integer>, CompletionStage<Done>> output = Sink.foreach(it -> log.info(it.toString()));

            SourceShape<Integer> inputShape = builder.add(input);
            FlowShape<Integer, Integer> incrementerShape = builder.add(incrementer);
            FlowShape<Integer, Integer> multiplierShape = builder.add(multiplier);
            SinkShape<Pair<Integer, Integer>> outputShape = builder.add(output);

            UniformFanOutShape<Integer, Integer> broadcast = builder.add(Broadcast.create(2));
            FanInShape2<Integer, Integer, Pair<Integer, Integer>> zip = builder.add(Zip.create());

            builder.from(inputShape).toFanOut(broadcast);

            builder.from(broadcast.out(0)).via(incrementerShape).toInlet(zip.in0());
            builder.from(broadcast.out(1)).via(multiplierShape).toInlet(zip.in1());
            builder.from(zip.out()).to(outputShape);

            return ClosedShape.getInstance();
        }));

        graph.run(system);
    }
}
